
package com.leo.jdkcore.kafka;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.utils.ShutdownableThread;

/**
 * @param <KType> kafka key selizable type
 * @param <VType> kafka value selizable type
 */
public class KafkaMsgConsumer<KType,VType> extends ShutdownableThread {
	protected static final Logger logger = LoggerFactory.getLogger(KafkaMsgConsumer.class);
	
	private final KafkaConsumer<KType, VType> consumer;
	private final String topic;
	private KafkaMsgHandler<KType,VType> dataHandler = null;
	private static AtomicInteger threadIndex = new AtomicInteger(1);
	
	//private Properties consumerProperties = null;
	
	
	public KafkaMsgConsumer(String topic,Properties props,Class<VType> vTypeClazz,Class<KType> kTypeClazz, KafkaMsgHandler<KType,VType> handler) {
		super("kafka-consumer-thread-"+threadIndex.getAndIncrement(), false);
		if(kTypeClazz.getName().equalsIgnoreCase(String.class.getName())){
			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
		}else if (kTypeClazz.getName().equalsIgnoreCase(Integer.class.getName())){
			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerDeserializer");
		}
		
		if(vTypeClazz.getName().equalsIgnoreCase(String.class.getName())){
			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
		}else{
			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArrayDeserializer");
		}
		consumer = new KafkaConsumer<KType,VType>(props);
		this.topic = topic;
		this.dataHandler = handler;
		//this.consumerProperties = props;
	}
	
	@Override
	public void doWork() {
		consumer.subscribe(Collections.singletonList(this.topic));
		if(dataHandler.canPoll()){
			ConsumerRecords<KType, VType> records = consumer.poll(2000);
			for (ConsumerRecord<KType, VType> record : records) {
				dataHandler.handlerMsg(record.value(),record);
				consumer.commitSync();
			}
		}else{
			logger.error("can't consumer records...because canpoll method return false...");
		}
	}
}